package top.cyuw.simplerpc.registry.zookeeper;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import top.cyuw.simplerpc.URL;
import top.cyuw.simplerpc.constant.RpcConstants;
import top.cyuw.simplerpc.registry.memory.MemoryServiceDiscovery;
import top.cyuw.simplerpc.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author chen
 * @date 2023/3/16 21:03
 */
@Slf4j
public class ZookeeperServiceDiscovery extends MemoryServiceDiscovery implements PathChildrenCacheListener {

    public ZookeeperServiceDiscovery() {
        try {
            CuratorFramework zkClient = ZookeeperClientUtils.createZkClient();
            List<String> nodes = ZookeeperClientUtils.getChildrenNodes(zkClient, ZookeeperClientUtils.ROOT_PATH);
            for (String nodeName : nodes) {
                loadServiceNodeFromZk(zkClient, ZookeeperClientUtils.ROOT_PATH + RpcConstants.URL_SEPARATOR + nodeName);
            }
            ZookeeperClientUtils.registerWatcher(zkClient, ZookeeperClientUtils.ROOT_PATH, this);
        } catch (Exception e) {
            log.error("load zk config to cache failed: " + e.getMessage(), e);
        }
    }

    @Override
    public void childEvent(CuratorFramework zkClient, PathChildrenCacheEvent event) throws Exception {
        PathChildrenCacheEvent.Type eventType = event.getType();
        if (eventType == PathChildrenCacheEvent.Type.CHILD_ADDED) {
            loadServiceNodeFromZk(zkClient, event.getData().getPath());
        } else if (eventType == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
            String value = new String(event.getData().getData(), StandardCharsets.UTF_8);
            URL url = URL.valueOf(value);
            getAddressList(url.getPath()).remove(url);
        }
    }

    private void loadServiceNodeFromZk(CuratorFramework zkClient, String path) throws Exception {
        String nodeData = ZookeeperClientUtils.getNodeData(zkClient, path);
        if (StringUtils.isNotEmpty(nodeData)) {
            URL url = URL.valueOf(nodeData);
            addAddress(url.getPath(), url);
        }
    }

}
